package com.gsm.wc;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 *  1 DataStream实现Wordcount：读socket（无界流）
 *
 *  2 并行度的优先级：
 *     代码：算子 > 全局env设置 > 提交时指定 > 配置文件
 *
 * @author gsm
 * @version 1.0
 */
public class WordCountStreamUnboundedDemo {

    public static void main(String[] args) throws Exception {

        // TODO 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // IDEA运行时，也可以看到webui，一般用于本地测试
        // 需要引入一个依赖 flink-runtime-web
        // 在idea运行，不指定并行度，默认就是 电脑的 线程数
//        StreamExecutionEnvironment env =
//                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1); // 全局设置

        // TODO 2. 读取数据： socket
        //  如果是centos  安装 yum install -y netcat 使用: nc -lk 7777
        //  如果是windows 安装  解压缩netcat-win32-1.12.zip  使用: netcat nc -lp 7777 或 nc64 -lp 7777
        //  D:\software\windows-software\netcat-win32-1.12\netcat-win32-1.12
        // 测试服务器启动flink + webUI
        //          也就是  [root@hadoop102 flink-1.17.0]# bin/start-cluster.sh
        //          再配合webUI时候 用hadoop102 别用localhost 否则还是连接不上
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
//        DataStreamSource<String> socketDS = env.socketTextStream("localhost", 7777);


        // TODO 3. 处理数据: 切换、转换、分组、聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                .flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word, 1));
                            }
                        }
                )
//                .setParallelism(2) // 可以给某个算子单独设置并行度 ,可以通过webUI查看
                .returns(Types.TUPLE(Types.STRING,Types.INT))
//                .returns(new TypeHint<Tuple2<String, Integer>>() {
//                    @Override
//                    public TypeInformation<Tuple2<String, Integer>> getTypeInfo() {
//                        return super.getTypeInfo(); // 默认就是super 所以这方法也可以省略 简化成如下写法
//                    }
//                })
//                .returns(new TypeHint<Tuple2<String, Integer>>() {})
                .keyBy(value -> value.f0)
                .sum(1);


        // TODO 4. 输出
        sum.print();

        // TODO 5. 执行
        env.execute();
    }
}

/**

 并行度的优先级：
    代码：算子 > 全局env设置 > 提交时指定 > 配置文件

 */